package com.sduept.component.message.core.config;

import com.sduept.kafka.KafkaTopicConfig;
import com.sduept.kafka.model.KafkaTopic;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Author: qcl
 * @Date: 2020/5/25 11:23
 * @Description: 动态 topic订阅
 */
@Configuration
@Slf4j
public class DynamicTopicsConfig implements InitializingBean {
    @Autowired
    private KafkaTopicConfig kafkaTopicConfig;
    @Override
    public void afterPropertiesSet() throws Exception {

        List<KafkaTopic> list = getKafkaTopicListByScopes(new String[]{"user"});
        String DynamicTopics = list.stream().map(e -> e.getName()).collect(Collectors.joining(","));
        System.setProperty("DynamicTopics",DynamicTopics);

    }

    /**
     * 获得指定范围的主题
     *
     * @param type
     * @return
     */
    private List<KafkaTopic> getKafkaTopicListByScopes(String[] scopes) {
        List<KafkaTopic> list = new ArrayList<KafkaTopic>();
        Map<String, KafkaTopic> topics = kafkaTopicConfig.getTopics();
        for (KafkaTopic t : topics.values()) {
            for (String scope : scopes) {
                if (!StringUtils.isEmpty(t.getScope()) && t.getScope().contains(scope))
                    list.add(t);
            }
        }
        return list;
    }
}
